package com.chis.netty.mqtt.client;

import com.chis.netty.mqtt.server.BrokerSessionHelper;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.handler.codec.mqtt.MqttFixedHeader;
import io.netty.handler.codec.mqtt.MqttMessage;
import io.netty.handler.codec.mqtt.MqttMessageFactory;
import io.netty.handler.codec.mqtt.MqttMessageType;
import io.netty.handler.codec.mqtt.MqttPublishVariableHeader;
import io.netty.handler.codec.mqtt.MqttQoS;
import io.netty.util.CharsetUtil;

/**
 * 
 * @Description 
 * @Company zwx
 * @author hhl
 * @version 2020年05月13日
 */
public class BrokerHandler extends SimpleChannelInboundHandler<MqttMessage> {


    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        String msg = "client message";
        byte[] bytes = msg.getBytes(CharsetUtil.UTF_8);
        ByteBuf buf = Unpooled.wrappedBuffer(bytes);

        BrokerSessionHelper.sendMessage(
                ctx,
                MqttMessageFactory.newMessage(
                        new MqttFixedHeader(MqttMessageType.PUBLISH, false, MqttQoS.AT_MOST_ONCE, false, 0),
                        new MqttPublishVariableHeader("topic_aa", 1),
                        buf),
                "INVALID",
                null,
                true);
    }

    protected void channelRead0(ChannelHandlerContext ctx, MqttMessage msg) throws Exception {
        switch (msg.fixedHeader().messageType())    {
            case CONNACK:
                System.out.println("CONNACK");
                // publishm(ctx, msg);
                break;
            case PUBLISH:
                System.out.println("PUBLISH");
                ByteBuf buf1 = Unpooled.wrappedBuffer((ByteBuf)msg.payload());
                System.out.println("接收服务端数据channelRead0: "+buf1.toString(CharsetUtil.UTF_8));
                break;
        }
       
    }


    private void publishm(ChannelHandlerContext ctx, MqttMessage msg) {
        ByteBuf buf1 = Unpooled.wrappedBuffer((ByteBuf)msg.payload());
        System.out.println("接收服务端数据channelRead0: "+buf1.toString(CharsetUtil.UTF_8));

        String msg2 = "server message";
        byte[] bytes2 = msg2.getBytes(CharsetUtil.UTF_8);
        ByteBuf buf = Unpooled.wrappedBuffer(bytes2);

        BrokerSessionHelper.sendMessage(
                ctx,
                MqttMessageFactory.newMessage(
                        new MqttFixedHeader(MqttMessageType.PUBLISH, false, MqttQoS.AT_MOST_ONCE, false, 0),
                        new MqttPublishVariableHeader("topic_aa", 1),
                        buf),
                "INVALID",
                null,
                true);
    }

}
